
#include <stdio.h>
#include <unistd.h>
#include <assert.h>
#include "xs_string.h"
#include "xs_malloc.h"
#include "xs_threadpool.h"


#define THD_NUM 15

static xs_pthread_mutex_t lock;
static unsigned exit_cnt;
static xs_thread_pool *g_threadpool_table;


//#define __DEBUG__

#ifdef __DEBUG__
#define DEBUG(format,...)	printf(format,##__VA_ARGS__)
#else
#define DEBUG(format,...) 
#endif

static void *work_thread(void *arg) 
{
	xs_thread_info *pTinfo = (xs_thread_info *) arg;
    	xs_thread_pool *threadpool_table = pTinfo->threadpool;

	//wake up waiting thread, notify it I am ready
    	xs_pthread_cond_signal(&threadpool_table->threadpool_cond);

    	while (!(threadpool_table->stop)) {
		if(pTinfo->func){
			DEBUG("thread %u is running\n", pTinfo->thread_id);
			pTinfo->func(pTinfo->arg);
			/* thread state shoulde be set idle after work */
			pTinfo->is_busy = FALSE;
			pTinfo->func = NULL;
			/* I am idle now */
            		xs_queue_enqueue_data(threadpool_table->idle, pTinfo);
		}

		/* wait cond for processing real job. */
		DEBUG("thread %u is waiting for a job\n", pTinfo->thread_id);
		xs_pthread_mutex_lock(&pTinfo->thread_mutex);
		xs_pthread_cond_wait(&pTinfo->thread_cond, &pTinfo->thread_mutex);
		xs_pthread_mutex_unlock(&pTinfo->thread_mutex);
		DEBUG("thread %u end waiting for a job\n", pTinfo->thread_id);

		if(pTinfo->threadpool->stop){
			DEBUG("thread %u stop\n", pTinfo->thread_id);
			break;
		}
	}
	DEBUG("Job done, thread %u is idle now.\n", pTinfo->thread_id);

	return NULL;
}


/**
 * member function reality. add new thread into the pool and run immediately.
 */
static xs_thread_info *xs_add_thread(xs_thread_pool *threadpool_table, xs_work func, void *arg)
{
	xs_thread_info *new_thread;

	xs_pthread_mutex_lock(&threadpool_table->threadpool_mutex);

	if (threadpool_table->maxinum <= threadpool_table->current){
		xs_pthread_mutex_unlock(&threadpool_table->threadpool_mutex);
		return NULL;
	}

	//malloc new thread info struct
	new_thread = threadpool_table->thread_info + threadpool_table->current;
	threadpool_table->current++;
	xs_pthread_mutex_unlock(&threadpool_table->threadpool_mutex);

	new_thread->threadpool = threadpool_table;
	//init new thread's cond & mutex
	xs_pthread_cond_init(&new_thread->thread_cond, NULL);
	xs_pthread_mutex_init(&new_thread->thread_mutex, NULL);

	//init status is busy, only new process job will call this function
	new_thread->is_busy = TRUE;
	new_thread->func = func;
	new_thread->arg = arg;
	new_thread->thread_id = xs_pthread_create(work_thread, new_thread);

	return new_thread;
}

/**
 * member function reality. delete idle thread in the pool.
 * only delete last idle thread in the pool.
 */
int xs_delete_thread(xs_thread_pool *threadpool_table)
{
	xs_thread_info *pThi;
	xs_thread_info tT;

	//current thread num can't < min thread num
	if (threadpool_table->current <= threadpool_table->mininum) return -1;

	//all threads are busy
	pThi = (xs_thread_info *) xs_queue_dequeue_data(threadpool_table->idle);
	if(!pThi) return -1;

	//after deleting idle thread, current thread num -1
	xs_pthread_mutex_lock(&threadpool_table->threadpool_mutex);
	threadpool_table->current--;
	/** swap this thread to the end, and free it! **/
	xs_memcpy(&tT, pThi, sizeof(xs_thread_info));
	xs_memcpy(pThi, threadpool_table->thread_info + threadpool_table->current, sizeof(xs_thread_info));
	xs_memcpy(threadpool_table->thread_info + threadpool_table->current, &tT, sizeof(xs_thread_info));
	xs_pthread_mutex_unlock(&threadpool_table->threadpool_mutex);

	//kill the idle thread and free info struct
	kill((pid_t)tT.thread_id, SIGKILL);
	xs_pthread_mutex_destroy(&tT.thread_mutex);
	xs_pthread_cond_destroy(&tT.thread_cond);

	return 0;
}


/**
 * member function reality. get current thread pool status:idle, normal, busy, .etc.
 */
int xs_get_thread_pool_status(xs_thread_pool *threadpool_table)
{
	float busy_num = 0.0;

	/* get busy thread number */
	busy_num = (threadpool_table->current - xs_queue_count(threadpool_table->idle));

	DEBUG("Current thread pool status, current num: %u, busy num: %u, idle num: %u\n", threadpool_table->current, (unsigned)busy_num, xs_queue_count(threadpool_table->idle));
	if(busy_num / (threadpool_table->current) < threadpool_table->busy_threshold)
		return 0;	/* idle status */
	else
		return 1;	/* busy or normal status */	
}

/**
 * internal interface. manage thread pool to delete idle thread.
 */
static void *xs_manage_thread(void *arg)
{
	xs_thread_pool *threadpool_table = (xs_thread_pool*) arg;//main thread pool struct instance

	//1?
	sleep(threadpool_table->manage_interval);

	do {
	    if (xs_get_thread_pool_status(threadpool_table) == 0) {
			do {
	            if (!xs_delete_thread(threadpool_table))
					break;
			} while (TRUE);
		}
		//1?
	    sleep(threadpool_table->manage_interval);
	} while (!threadpool_table->stop);

	return NULL;
}

float xs_get_busy_threshold(xs_thread_pool *threadpool_table)
{
    	return threadpool_table->busy_threshold;
}

int xs_set_busy_threshold(xs_thread_pool *threadpool_table, float bt)
{
	if(bt <= 1.0 && bt > 0.)
		threadpool_table->busy_threshold = bt;
	
    	return (threadpool_table->busy_threshold);
}

unsigned xs_get_manage_interval(xs_thread_pool *threadpool_table)
{
	return threadpool_table->manage_interval;
}

int xs_set_manage_interval(xs_thread_pool *threadpool_table, unsigned mi)
{
	threadpool_table->manage_interval = mi;

	return (threadpool_table->manage_interval);
}

static inline int xs_threadpool_init(xs_thread_pool *threadpool_table)
{
	int i;
	xs_thread_info *pThi;

	if(threadpool_table == NULL) return (-1);
	
    	threadpool_table->idle = xs_queue_create();
    	threadpool_table->stop = FALSE;
    	threadpool_table->busy_threshold = BUSY_THRESHOLD;
    	threadpool_table->manage_interval = MANAGE_INTERVAL;

	//create work thread and init work thread info
    	for (i = 0; i < threadpool_table->mininum; i++) {

        	pThi = threadpool_table->thread_info + i;
        	pThi->threadpool = threadpool_table;
		pThi->is_busy = FALSE;
		pThi->func = NULL;
		pThi->arg = NULL;
		xs_pthread_cond_init(&pThi->thread_cond, NULL);
		xs_pthread_mutex_init(&pThi->thread_mutex, NULL);

        	xs_queue_enqueue_data(threadpool_table->idle, pThi);

		pThi->thread_id = xs_pthread_create(work_thread, pThi);
		if(pThi->thread_id == 0){	
			/* thread create error */
            		xs_queue_destroy(threadpool_table->idle);
			return -1;
		}
	}

    	threadpool_table->manage_thread = xs_pthread_create(xs_manage_thread, threadpool_table);
    	if(threadpool_table->manage_thread == 0){
        	xs_queue_destroy(threadpool_table->idle);
		return 0;
	}
	printf("_________%d  threadpool_table->current %d\n", i, threadpool_table->current);

	//wait for all threads are ready
    	while(i++ < threadpool_table->current){
        	xs_pthread_mutex_lock(&threadpool_table->threadpool_mutex);
        	xs_pthread_cond_wait(&threadpool_table->threadpool_cond, &threadpool_table->threadpool_mutex);
        	xs_pthread_mutex_unlock(&threadpool_table->threadpool_mutex);
	}

	DEBUG("All threads are ready now\n");

	return 0;
}

/**
	min is a initialization thread number when call xs_threadpool_create,
	max is a initialization value, too. But value of max stands for how many
	threads(up limits value) does the threadpool surpport. When someone 
	starts task count larger than max, only 0~max tasks can be scheduled.
*/
xs_thread_pool *xs_threadpool_create(unsigned min, unsigned max)
{
	xs_thread_pool *threadpool_table;

	if( min > max ) return NULL;
	
	threadpool_table = (xs_thread_pool*) xs_malloc(sizeof(xs_thread_pool));

	xs_memset(threadpool_table, 0, sizeof(xs_thread_pool));
	threadpool_table->mininum = min;
	threadpool_table->current = min;
	threadpool_table->maxinum = max;
	threadpool_table->exitted = 0;	/* No thread exitted */
	xs_pthread_mutex_init(&threadpool_table->threadpool_mutex, NULL);
	xs_pthread_cond_init(&threadpool_table->threadpool_cond, NULL);
	xs_pthread_mutex_init(&threadpool_table->threadpool_exit_mutex, NULL);
	xs_pthread_cond_init(&threadpool_table->threadpool_exit_cond, NULL);

	if (NULL != threadpool_table->thread_info)
		xs_free(threadpool_table->thread_info);

	threadpool_table->thread_info = (xs_thread_info*) xs_malloc(sizeof(xs_thread_info) * threadpool_table->maxinum);
	xs_memset(threadpool_table->thread_info, 0, sizeof(xs_thread_info) * threadpool_table->maxinum);

	exit_cnt = 0;
	xs_pthread_mutex_init(&lock, NULL); /* init exit counter mutex */

	return ((xs_threadpool_init(threadpool_table) < 0) ? NULL:threadpool_table);
}
void xs_threadpool_run(xs_thread_pool *threadpool_table)
{
	/* Locking threadpool_exit_mutex before entries xs_pthread_cond_wait */
	xs_pthread_mutex_lock(&threadpool_table->threadpool_exit_mutex);
	xs_pthread_cond_wait(&threadpool_table->threadpool_exit_cond, &threadpool_table->threadpool_exit_mutex);
	xs_pthread_mutex_unlock(&threadpool_table->threadpool_exit_mutex);
	printf("------------ out of xs_pthread_run ----------!\n");
}
static inline void xs_threadpool_exit(xs_thread_pool *threadpool_table)
{
    	xs_pthread_cond_signal(&threadpool_table->threadpool_exit_cond);
}

/**
 * member function reality. thread pool entirely close function.
 */
void xs_threadpool_close(xs_thread_pool *threadpool_table, BOOL wait)
{
	int i;

    	threadpool_table->stop = TRUE;

	if ( wait) {
		/* Wait for main thread(main) end, so we need join all of the threads to thread pool */
		DEBUG("current number of threads: %u", threadpool_table->current);
		for (i = 0; i < threadpool_table->current; i++) {
			xs_pthread_cond_signal(&threadpool_table->thread_info[i].thread_cond);
		}
			
		for (i = 0; i < threadpool_table->current; i++) {
			/* join all the thread to quit */
			xs_pthread_join(threadpool_table->thread_info[i].thread_id, NULL);
			xs_pthread_mutex_destroy(&threadpool_table->thread_info[i].thread_mutex);
			xs_pthread_cond_destroy(&threadpool_table->thread_info[i].thread_cond);
		}
		
	} else {
		/* close work thread */
            	for (i = 0; i < threadpool_table->current; i++) {
            		kill((pid_t)threadpool_table->thread_info[i].thread_id, SIGKILL);
			xs_pthread_mutex_destroy(&threadpool_table->thread_info[i].thread_mutex);
            		xs_pthread_cond_destroy(&threadpool_table->thread_info[i].thread_cond);
		}
	}
	
	/* Here close management thread */
	kill((pid_t)threadpool_table->manage_thread, SIGKILL);
	
	xs_pthread_mutex_destroy(&threadpool_table->threadpool_mutex);
	xs_pthread_cond_destroy(&threadpool_table->threadpool_cond);
	xs_pthread_mutex_destroy(&threadpool_table->threadpool_exit_mutex);
	xs_pthread_cond_destroy(&threadpool_table->threadpool_exit_cond);
	xs_queue_destroy(threadpool_table->idle);

	xs_free(threadpool_table->thread_info);
	
    	threadpool_table->thread_info = NULL;
	
}

/** 
*	Main thread quit after all of threads finished there job.
*	So, when run task, this macro define must be called at the last.
*/
#define __xs_check_thread_terminal__(threadpool_table)	{\
										xs_pthread_mutex_lock(&lock);\
										exit_cnt++;\
										xs_pthread_mutex_unlock(&lock);\
										printf("exit number : %d\n", exit_cnt);\
										if(exit_cnt == threadpool_table->tasks){\
											exit_cnt = 0;\
											xs_threadpool_exit(threadpool_table);\
										}\
									}
void xs_work_fun(void *arg)
{
	int i;
	int idx = (*(int *) (arg));

	i = 1000000.0 + (int)(9000000.0 * rand() / RAND_MAX);

	//while(1)	
	{
		
		fprintf(stderr, "Thread %u, begin: job %d, sleep %d us \n...\n", pthread_self(), idx, i);
		usleep(i);
		fprintf(stderr, "End:   job %d\n", idx);
		
	}
	
	__xs_check_thread_terminal__(g_threadpool_table)
}
int _xs_work(xs_thread_pool *threadpool_table, xs_work func, void *arg)
{
    	xs_thread_info *pThi ;

    	pThi = (xs_thread_info *) xs_queue_dequeue_data(threadpool_table->idle);
	if(pThi){
		DEBUG("Fetch a thread from pool.\n");
		pThi->is_busy = TRUE;
        	pThi->func = func;
		pThi->arg = arg;
		//let the thread to deal with this job
		DEBUG("wake up thread %u\n", pThi->thread_id);
		xs_pthread_cond_signal(&pThi->thread_cond);
	} else {
			//if all current thread are busy, new thread is created here
	            	if(!(pThi = xs_add_thread(threadpool_table, func, arg))){
				printf("The thread pool is full, no more thread available.\n");
				return -1;
			}
			/* should I wait? */
			//xs_pthread_mutex_lock(&threadpool_table->threadpool_mutex);
			//xs_pthread_cond_wait(&threadpool_table->threadpool_cond, &threadpool_table->threadpool_mutex);
			//xs_pthread_mutex_unlock(&threadpool_table->threadpool_mutex);

			printf("No more idle thread, a new thread is created.\n");
	}
	threadpool_table->tasks++;
	return 0;
}

int xs_threadpool_entry(int argc, char **argv)
{
	argc = argc;
	argv = argv;
	int i;
	
    	g_threadpool_table = xs_threadpool_create(1, 3);

	assert(g_threadpool_table);
	
	srand((int)time(0));

	for(i=1; i < THD_NUM; i++)
		_xs_work(g_threadpool_table, xs_work_fun, &i);
	
    	xs_threadpool_run(g_threadpool_table);

    	xs_threadpool_close(g_threadpool_table, TRUE);
		
	xs_free(g_threadpool_table);

	fprintf(stderr, "All threads exitted, main thread quitting... \n");
	
	return 0;
}
